package com.ws.datasync.service.impl;

import com.ws.datasync.common.util.Result;
import com.ws.datasync.service.CompareDataService;
import org.apache.commons.collections4.ListUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class CompareDataServiceImpl implements CompareDataService {

    private static final Logger logger = LoggerFactory.getLogger(CompareDataServiceImpl.class);

    @Autowired
    @Qualifier("primaryJdbcTemplate")
    protected JdbcTemplate targetJdbcTemplate;

    @Autowired
    @Qualifier("secondaryJdbcTemplate")
    protected JdbcTemplate sourceJdbcTemplate;

    @Value("${spring.datasource.primary.name}")
    public String targetDbName;

    @Value("${spring.datasource.secondary.name}")
    public String sourceDbName;

    @Value("${spring.datasource.primary.type}")
    public String targetDbType;

    @Value("${spring.datasource.secondary.type}")
    public String sourceDbType;

    @Value("${business.ignore-size}")
    private int ignoreSize;

    private RowMapper<Map.Entry<String, String>> rm = (rs, rowNum) -> {
        String tableName = rs.getString("tableName");
        String primaryKey = rs.getString("primaryKey");
        return new AbstractMap.SimpleEntry<>(tableName, primaryKey);
    };

    @Override
    public Result comparePrimaryData() {
        logger.info("开始对比数据");
        String selectPrimaryKeySql = "SELECT c.TABLE_NAME AS 'tableName', c.column_name AS 'primaryKey' " +
                "FROM information_schema.`COLUMNS` c " +
                "WHERE c.table_schema = '&1' AND c.COLUMN_KEY = 'PRI'";

        Map<String, String> sourceTablePrimaryMap = sourceJdbcTemplate.query(selectPrimaryKeySql.replace("&1", sourceDbName), rm)
                .stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        logger.info("数据库表主键信息，dbName:{}，primaryKeys:{}", sourceDbName, sourceTablePrimaryMap);
        Map<String, String> targetTablePrimaryMap = targetJdbcTemplate.query(selectPrimaryKeySql.replace("&1", targetDbName), rm)
                .stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        logger.info("数据库表主键信息，dbName:{}，primaryKeys:{}", targetDbName, targetTablePrimaryMap);
        if (CollectionUtils.isEmpty(sourceTablePrimaryMap) || CollectionUtils.isEmpty(targetTablePrimaryMap)) {
            return Result.error("获取主键字段名称为空");
        }
        try {
            targetTablePrimaryMap.forEach((tableName, primaryKey) -> {
                if (sourceTablePrimaryMap.containsKey(tableName)) {
                    String sourcePrimaryKey = sourceTablePrimaryMap.get(tableName);
                    String targetPrimaryKey = targetTablePrimaryMap.get(tableName);
                    diffTablePrimaryValue(tableName, sourcePrimaryKey, targetPrimaryKey);
                }
            });
        } catch (Exception e) {
            logger.error("对比数据异常", e);
            return Result.error("对比数据异常");
        }
        logger.info("结束对比数据");
        return Result.success();
    }


    @Override
    public Result comparePrimaryDataAsync() {
        logger.info("开始对比数据");
        String selectPrimaryKeySql = "SELECT c.TABLE_NAME AS 'tableName', c.column_name AS 'primaryKey' " +
                "FROM information_schema.`COLUMNS` c " +
                "WHERE c.table_schema = '&1' AND c.COLUMN_KEY = 'PRI'";

        Map<String, String> sourceTablePrimaryMap = sourceJdbcTemplate.query(selectPrimaryKeySql.replace("&1", sourceDbName), rm)
                .stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        logger.info("数据库表主键信息，dbName:{}，primaryKeys:{}", sourceDbName, sourceTablePrimaryMap);
        Map<String, String> targetTablePrimaryMap = targetJdbcTemplate.query(selectPrimaryKeySql.replace("&1", targetDbName), rm)
                .stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        logger.info("数据库表主键信息，dbName:{}，primaryKeys:{}", targetDbName, targetTablePrimaryMap);
        if (CollectionUtils.isEmpty(sourceTablePrimaryMap) || CollectionUtils.isEmpty(targetTablePrimaryMap)) {
            return Result.error("获取主键字段名称为空");
        }

        ThreadPoolExecutor executor = new ThreadPoolExecutor(10,
                10, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(500));

        targetTablePrimaryMap.forEach((tableName, primaryKey) -> {
            if (sourceTablePrimaryMap.containsKey(tableName)) {
                executor.submit(() -> {
                    String sourcePrimaryKey = sourceTablePrimaryMap.get(tableName);
                    String targetPrimaryKey = targetTablePrimaryMap.get(tableName);
                    diffTablePrimaryValue(tableName, sourcePrimaryKey, targetPrimaryKey);
                });
            }
        });
        executor.shutdown();
        logger.info("结束对比数据");
        return Result.success();
    }

    @Override
    public Result diffData(String tableName) {
        logger.info("开始对比数据");
        String selectPrimaryKeySql = "SELECT c.column_name AS 'primaryKey' " +
                "FROM information_schema.`COLUMNS` c " +
                "WHERE c.table_schema = ? AND c.COLUMN_KEY = 'PRI' AND c.TABLE_NAME=? ";
        try {

         String sourcePrimaryKey = sourceJdbcTemplate.queryForObject(selectPrimaryKeySql, new Object[]{sourceDbName,tableName},String.class);
         String targetPrimaryKey=targetJdbcTemplate.queryForObject(selectPrimaryKeySql,new Object[]{targetDbName,tableName},String.class);
         diffTablePrimaryValue( tableName,  sourcePrimaryKey,  targetPrimaryKey);
        } catch (Exception e){
            e.printStackTrace();
            return Result.error("系统错误");
        }
        return Result.success("成功差分");
    }

    @Override
    public Result syncDiffData(String tableName) {
        return null;
    }

    private void diffTablePrimaryValue(String tableName, String sourcePrimaryKey, String targetPrimaryKey) {
        String sourceSql = "SELECT `" + sourcePrimaryKey + "` FROM " + tableName;
        String targetSql = "SELECT `" + targetPrimaryKey + "` FROM " + tableName;
        List<String> sourceList = sourceJdbcTemplate.queryForList(sourceSql, String.class);
        List<String> targetList = targetJdbcTemplate.queryForList(targetSql, String.class);
        // 转为map用于提高获取差集的效率
        HashMap<String, String> targetMap = targetList.stream().collect(Collectors.toMap(s -> s, s -> s, (s1, s2) -> s1, HashMap::new));
        // 记录不存在的值
        List<Object[]> notExistList = new ArrayList<>();
        sourceList.forEach(s -> {
            if (!targetMap.containsKey(s)) {
                notExistList.add(new Object[]{tableName, sourcePrimaryKey, s});
            }
        });
        logger.info("表{},主键值{},在目标数据库中不存在有{}个", tableName, sourcePrimaryKey, notExistList.size());
        if (notExistList.size() > 0 && notExistList.size() < ignoreSize) {
            // 清除表数据
            String deleteSql = "DELETE FROM primary_data_diff WHERE table_name= '" + tableName + "'";
            targetJdbcTemplate.update(deleteSql);
            // 分批插入primary_data_diff表
            String sql = "INSERT INTO primary_data_diff (table_name, primary_key, primary_value) VALUES (?, ?, ?)";
            List<List<Object[]>> subLists = ListUtils.partition(notExistList, getLimit(notExistList.size()));
            subLists.forEach(subList -> {
                targetJdbcTemplate.batchUpdate(sql, subList);
            });
        } else {
            // TODO 超过100w条的数据只记录表名
            logger.info("表{},主键值{},在目标数据库中不存在有{}个，超过阈值，不记录", tableName, sourcePrimaryKey, notExistList.size());
        }
    }

    private int getLimit(int size) {
        if (size > 1000000) {
            return 50000;
        } else if (size > 500000) {
            return 20000;
        } else if (size > 100000) {
            return 10000;
        } else {
            return 5000;
        }
    }


}
